package main

import (
	"bytes"
	"log"
	"time"

	"fmt"
	"github.com/streadway/amqp"
)

func failOnError(err error, msg string) {
	if err != nil {
		log.Fatalf("%s: %s", msg, err)
	}
}

type ReceivedRabbitMQ struct {
	conn    *amqp.Connection     //连接
	ch      *amqp.Channel        //频道
	q       amqp.Queue           //队列
	msgchan <-chan amqp.Delivery //管道
}

func NewReceivedRabbitMQ(url string, queuename string) *ReceivedRabbitMQ {
	myrecive := new(ReceivedRabbitMQ)
	var err error
	myrecive.conn, err = amqp.Dial(url)
	failOnError(err, "Failed to connect to RabbitMQ")

	myrecive.ch, err = myrecive.conn.Channel()
	failOnError(err, "Failed to open a channel")

	q, err := myrecive.ch.QueueDeclare(
		queuename, // name
		true,      // durable
		false,     // delete when unused
		false,     // exclusive
		false,     // no-wait
		nil,       // arguments
	)
	failOnError(err, "Failed to declare a queue")

	err = myrecive.ch.Qos(
		1,     // prefetch count
		0,     // prefetch size
		false, // global
	)
	failOnError(err, "Failed to set QoS")

	myrecive.msgchan, err = myrecive.ch.Consume(
		q.Name, // queue
		"",     // consumer
		false,  // auto-ack
		false,  // exclusive
		false,  // no-local
		false,  // no-wait
		nil,    // args
	)
	failOnError(err, "Failed to register a consumer")
	return myrecive
}
func (rr *ReceivedRabbitMQ) ReceiveData(ServerMsg func(msg string)) {
	forever := make(chan bool)

	go func() {
		for d := range rr.msgchan {
			ServerMsg(fmt.Sprintf("%s", d.Body))
			log.Printf("Received a message: %s", d.Body)
			dot_count := bytes.Count(d.Body, []byte("."))
			t := time.Duration(dot_count)
			time.Sleep(t * time.Second)
			log.Printf("Done")
			d.Ack(false)
		}
	}()

	log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
	<-forever
}
func (rr *ReceivedRabbitMQ) Close() {
	rr.ch.Close()
	rr.conn.Close()
}
func (rr *ReceivedRabbitMQ) ReceiveDataTimeOut(s int) []string {
	//forever := make(chan bool)
	returnlist := []string{} //返回的字符串数组
	wait := make(chan string)

	go func() {
		for d := range rr.msgchan {
			//ServerMsg(fmt.Sprintf("%s",d.Body))
			log.Printf("Received a message: %s", d.Body)
			dot_count := bytes.Count(d.Body, []byte("."))
			t := time.Duration(dot_count)
			time.Sleep(t * time.Second)
			log.Printf("Done")
			returnlist = append(returnlist, fmt.Sprintf("%s", d.Body)) //收到的数据加入数组
			d.Ack(false)
		}
	}()
	select {
	case res := <-wait:
		fmt.Println(res)
	case <-time.After(time.Second * time.Duration(s)):
		return returnlist
	}

	return []string{}
}
